package indi.liangli.zookeeper.lock;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.List;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

public class DistributedLock implements Lock, Watcher {

    private ZooKeeper zk = null;
    private String ROOT_NODE = "/locks";
    private String CURRENT_NODE;//获取锁的节点
    private String WAIT_NNODE;//等待的节点

    //用来做等待和唤醒的
    private CountDownLatch countDownLatch = null;

    public DistributedLock() {
        try {
            zk = new  ZooKeeper("192.168.78.101:2181",4000,this) ;
            Stat stat = zk.exists(ROOT_NODE, false);//不注册监听
            if(stat == null){
                //创建一个持久化节点
                zk.create(ROOT_NODE,"0".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }


    @Override
    public boolean tryLock() {
        try {
            CURRENT_NODE = zk.create(ROOT_NODE+"/" ,"0".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);
            System.out.println(Thread.currentThread().getName()+"-->"+CURRENT_NODE+" ,尝试获得锁");
            List<String> children = zk.getChildren(ROOT_NODE, false);
            SortedSet<String> sortedSet = new TreeSet<>();
            children.parallelStream().forEach(path ->{
               sortedSet.add(ROOT_NODE+"/"+path);
            });
            String first = sortedSet.first();
            if(CURRENT_NODE.equalsIgnoreCase(first)){//如果当前节点是最小的节点，表示获得锁
                System.out.println(Thread.currentThread().getName()+"-->"+CURRENT_NODE+" 获得锁成功");
                return true;
            }
            SortedSet<String> lessThenMe = ((TreeSet<String>) sortedSet).headSet(CURRENT_NODE);
            //如果有比当前节点更小的节点，设置当前节点监听上一个节点
            if(!lessThenMe.isEmpty()){
                WAIT_NNODE = lessThenMe.last();
            }
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return false;
    }


    @Override
    public void lock() {

        try {
            if(this.tryLock()){
                return;
            }
            waitForLock(WAIT_NNODE);
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

    private void waitForLock(String prev) throws KeeperException, InterruptedException {
        Stat stat = zk.exists(prev, true);
        if(stat != null){//表示当前节点确实存在上一个节点
            System.out.println(Thread.currentThread().getName()+"-> "+" 等待锁 "+prev+"释放");
            this.countDownLatch = new CountDownLatch(1);
            countDownLatch.await();

        }


    }

    @Override
    public void lockInterruptibly() throws InterruptedException {

    }


    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return false;
    }

    @Override
    public void unlock() {
        try {
            //-1 表示强制删除
            zk.delete(CURRENT_NODE,-1);
            zk.close();
            CURRENT_NODE = null;
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    @Override
    public Condition newCondition() {
        return null;
    }

    @Override
    public void process(WatchedEvent event) {
        //事件监听触发回调
        if(this.countDownLatch != null){
            countDownLatch.countDown();
            System.out.println("event "+event.getType()+"path"+event.getPath());
//            CURRENT_NODE = event.getPath();
        }
    }

    public static void main(String[] args) throws IOException {
        CountDownLatch countDownLatch = new CountDownLatch(10);
        for(int i=0;i<10;i++){
            new Thread(()->{
                try {
                    countDownLatch.await();
                    DistributedLock distributedLock = new DistributedLock();
                    distributedLock.lock();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            },"thread："+i).start();
            //模拟10个线程并发
            countDownLatch.countDown();
        }
        System.in.read();
    }
}
